package org.lzfto.flink.demo.business.flinkTest.flinkObjects.source.clickdemo;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import org.apache.flink.api.connector.source.Boundedness;
import org.apache.flink.api.connector.source.Source;
import org.apache.flink.api.connector.source.SourceReader;
import org.apache.flink.api.connector.source.SourceReaderContext;
import org.apache.flink.api.connector.source.SplitEnumerator;
import org.apache.flink.api.connector.source.SplitEnumeratorContext;
import org.apache.flink.core.io.SimpleVersionedSerializer;
import org.lzfto.flink.demo.business.flinkTest.entity.ClickEvent;
import org.lzfto.flink.demo.business.flinkTest.entity.ClickEventSourceConfig;
import org.lzfto.flink.demo.business.flinkTest.flinkObjects.source.CustEnumeratorCKT;
import org.lzfto.flink.demo.business.flinkTest.flinkObjects.source.JsonSimpleVersionedSerializer;

/**
 * 模拟生成点击时间流。 存在一个开始时间和结束时间
 * @author lzfto
 * @created 2025年7月9日 11:23:09
 */
public class ClickEventSource implements Source<ClickEvent,ClickEventSplit,CustEnumeratorCKT> {
	private static final long serialVersionUID = 1L;
	private final ClickEventSourceConfig config;
	
	public ClickEventSource(ClickEventSourceConfig config) {
		this.config=config;
	}

	@Override
	public SourceReader<ClickEvent, ClickEventSplit> createReader(SourceReaderContext readerContext) throws Exception {
		ClickEventSourceReader reader=new ClickEventSourceReader(config);
		return reader;
	}

	@Override
	public Boundedness getBoundedness() {
		//这是一个有界限的流量测试
		return Boundedness.BOUNDED;
	}

	@Override
	public SplitEnumerator<ClickEventSplit, CustEnumeratorCKT> createEnumerator(
			SplitEnumeratorContext<ClickEventSplit> enumContext) throws Exception {
		List<ClickEventSplit> initialSplits = new ArrayList<>();
		//创建4个分片
		initialSplits.add(new ClickEventSplit("A01"));
		//initialSplits.add(new ClickEventSplit("A02"));
		//initialSplits.add(new ClickEventSplit("A03"));
		//initialSplits.add(new ClickEventSplit("A04"));
		return new ClickEventEnumerator(enumContext, initialSplits);
	}

	@Override
	public SplitEnumerator<ClickEventSplit, CustEnumeratorCKT> restoreEnumerator(
			SplitEnumeratorContext<ClickEventSplit> enumContext, CustEnumeratorCKT checkpoint) throws Exception {
		
		List<ClickEventSplit> splits = new ArrayList<>();
		for (byte[] serializedSplit : checkpoint.getSerializedAllSplits()) {
			splits.add(getSplitSerializer().deserialize(1, serializedSplit));
		}
		ClickEventEnumerator enumerator = new ClickEventEnumerator(enumContext, splits);
		
		// 恢复待分配的分片
		for (Map.Entry<Integer, List<byte[]>> entry : checkpoint.getSerializedPendingSplits().entrySet()) {
			List<ClickEventSplit> pendingSplits = new ArrayList<>();
			for (byte[] serializedSplit : entry.getValue()) {
				pendingSplits.add(getSplitSerializer().deserialize(1, serializedSplit));
			}
			enumerator.addSplitsBack(pendingSplits, entry.getKey());
		}
		
		return enumerator;
	}

	@Override
	public SimpleVersionedSerializer<ClickEventSplit> getSplitSerializer() {
		return new ClickEventSplitSerializer();
	}
	
	private static class ClickEventSplitSerializer implements SimpleVersionedSerializer<ClickEventSplit> {
        @Override
        public int getVersion() {
            return 1;
        }

        @Override
        public byte[] serialize(ClickEventSplit split) {
            return split.splitId().getBytes();  
        }

        @Override
        public ClickEventSplit deserialize(int version, byte[] serialized) {
            String id=new String(serialized);
            return new ClickEventSplit(id);
        }
    }

	@Override
	public SimpleVersionedSerializer<CustEnumeratorCKT> getEnumeratorCheckpointSerializer() {
		return new JsonSimpleVersionedSerializer();
	}
     
}